package com.shengzai.flink.core

import com.alibaba.fastjson2.{JSON, JSONObject}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

import java.lang
import java.time.Duration

object Demo6EventTime {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val socketDS: DataStream[String] = env.socketTextStream("master", 8888)

    val mapDS: DataStream[(String,Long)] = socketDS
      .map(
        line => {
          val split: Array[String] = line.split(",")
          (split(0),split(1).toLong)
        }
      )

    val assDS: DataStream[(String, Long)] = mapDS.assignTimestampsAndWatermarks(
      WatermarkStrategy
        //指定水位线前移的时间（数据最大乱序的时间）
        .forBoundedOutOfOrderness[(String, Long)](Duration.ofSeconds(5))
        //指定时间字段
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
          override def extractTimestamp(t: (String, Long), l: Long): Long = t._2
        })
    )

    assDS
      .map(line =>(line._1,1))
      .keyBy(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))
      .sum(1)
      .print()



    env.execute()


  }

}
